package com.example.hbase_server.controller;


import com.example.hbase_server.service.Func;
import net.minidev.json.JSONObject;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

import java.io.IOException;
import java.math.BigInteger;
import java.util.Map;


@RestController
@RequestMapping("/api")
public class Api {

    @Autowired
    Func func;

    @ModelAttribute
    public void settingBaseUrl() {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment _tnv = StreamTableEnvironment.create(env);
    }


    @PostMapping(path = "/addComment")
    public Map<String, Object> addComment(@RequestBody Map<String, String> DataInfo ) throws Exception {
        return func.addComment(DataInfo.get("userId"), DataInfo.get("movieId"), Double.parseDouble(DataInfo.get("rating")));
    }

    @PostMapping(path = "/getRating")
    public Map<String, Object> getRating(@RequestBody Map<String, String> DataInfo ) throws Exception {
        return func.getRating(DataInfo.get("movieId"));
    }

    @PostMapping(path = "/getMovies")
    public Map<String, Object> getMovies(@RequestBody Map<String, String> DataInfo ) throws Exception {
        return func.getMovies(DataInfo.get("startRow"),DataInfo.get("pageSize"));
    }

    @PostMapping(path = "/searchMovieRating")
    public Map<String, Object> searchMovieRating(@RequestBody Map<String, String> DataInfo ) throws Exception {
        return func.searchMovieRating(DataInfo.get("name"));
    }

    @GetMapping(path = "/getTenMovie")
    public Map<String, Object> getTenMovie() throws Exception {
        return func.getTenMovie();
    }
}
